use std::sync::{mpsc, Arc};
use std::thread;

use futures::{select, FutureExt};
use honeybee::prelude::*;
use honeybee::Result;

use crate::config::CONFIG;
use crate::listen_delegate;
use crate::{apps, gmsg, umsg, users};

pub fn run(shutdown_tx: mpsc::SyncSender<&'static str>) -> Addr<ShutdownActor> {
    let (tx, rx) = mpsc::sync_channel(1);
    let _ = thread::Builder::new()
        .name("async_main_thread".into())
        .spawn(move || {
            block_on(async move {
                let shutdown_addr = honeybee::prelude::init().await.unwrap();
                tx.send(shutdown_addr.clone()).unwrap();
                select! {
                    result = async_main().fuse() => {
                        if let Err(err) = result {
                            error!("async_main err:{}", err);
                        }
                    }
                    _ = shutdown_addr.wait_for_stop().fuse() => {}
                }
            });
            info!("async_main thread stoped");
            let _ = shutdown_tx.send("async_main exit");
        });
    rx.recv().unwrap()
}

async fn async_main() -> Result<()> {
    etcd::connect(&CONFIG.server.etcd_endpoints, None)
        .await
        .map_err(|err| {
            anyhow!(
                "connect etcd server:{:?} failed:{}",
                CONFIG.server.etcd_endpoints,
                err
            )
        })?;

    info!("try wait init mongo:{} ...", CONFIG.mongodb.url);
    let client = honeybee::mongodb::Client::with_uri_str(&CONFIG.mongodb.url)
        .await
        .map_err(|err| {
            anyhow!(
                "connect mongodb server:{} failed:{}",
                CONFIG.mongodb.url,
                err
            )
        })?;
    client.list_database_names(None, None).await?;
    umsg_db::set_client(client.clone());
    gmsg_db::set_client(client);

    let _ = TcpListenerActor::new(
        &CONFIG.node.node_type,
        &CONFIG.node.node_name,
        CONFIG.server.listen_port,
        proto::get_message_register(),
        Some(Arc::new(listen_delegate::Delegate)),
    )
    .start()
    .await?;

    let _ = etcd_usc_server::EtcdUscServerActor::new(
        &CONFIG.node.node_name,
        CONFIG.server.node_total_number,
        format!("{}:{}", CONFIG.server.service_ip, CONFIG.server.listen_port),
    )?
    .start()
    .await?;

    let _ = apps::Apps::from_registry().await?;
    let _ = users::Users::new().await.start().await?;
    let _ = umsg::UmsgActor::from_registry().await?;
    let _ = gmsg::GmsgActor::from_registry().await?;

    futures::future::pending::<()>().await;
    Ok(())
}
